package org.jianfa.mq.seckill.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.jianfa.mq.seckill.entity.po.OrderInfo;
import org.jianfa.mq.seckill.redis.RedisService;
import org.jianfa.mq.seckill.service.OrderService;
import org.jianfa.mq.seckill.service.SeckillService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

@Service
public class RocketMqConsumer {

    private static final Logger logger = LoggerFactory.getLogger(RocketMqConsumer.class);

    private DefaultMQPushConsumer consumer1;

    private DefaultMQPushConsumer consumer2;

    @Autowired
    private SeckillService seckillService;

    @Autowired
    private OrderService orderService;

    @PostConstruct
    public void postConstrct() throws Exception{
        consumer1 = new DefaultMQPushConsumer("SeckillConsumer");
        consumer1.setNamesrvAddr("127.0.0.1:9876");
        /*
         * Specify where to start in case the specified consumer group is a brand new one.
         */
        consumer1.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        /*
         * Subscribe one more more topics to consume.
         */
        consumer1.subscribe("Seckill", "*");

        /*
         *  Register callback to execute on arrival of messages fetched from brokers.
         */
        consumer1.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        byte[] body = msg.getBody();
                        String message = new String(body, RemotingHelper.DEFAULT_CHARSET);
                        SeckillMessage seckillMessage = RedisService.stringToBean(message, SeckillMessage.class);
                        seckillService.seckill(seckillMessage);
                    } catch (Exception ex) {
                        logger.error("消费消息失败", ex);
                    }
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        /*
         *  Launch the consumer instance.
         */
        consumer1.start();

        consumer2 = new DefaultMQPushConsumer("OrderInfoConsumer");
        consumer2.setNamesrvAddr("127.0.0.1:9876");
        /*
         * Specify where to start in case the specified consumer group is a brand new one.
         */
        consumer2.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        /*
         * Subscribe one more more topics to consume.
         */
        consumer2.subscribe("OrderInfo", "*");

        /*
         *  Register callback to execute on arrival of messages fetched from brokers.
         */
        consumer2.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        byte[] body = msg.getBody();
                        String message = new String(body, RemotingHelper.DEFAULT_CHARSET);
                        OrderInfo orderInfo = RedisService.stringToBean(message, OrderInfo.class);
                        orderService.checkOrder(orderInfo);
                    } catch (Exception ex) {
                        logger.error("消费消息失败", ex);
                    }
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        /*
         *  Launch the consumer instance.
         */
        consumer2.start();
    }

    @PreDestroy
    public void preDestroy() {
        consumer1.shutdown();
    }
}
